package org.leiyang.service.impls;

import org.leiyang.common.dtos.ConnectedMsgServerInfo;
import org.leiyang.common.dtos.MsgServerInfo;
import org.leiyang.common.dtos.R;
import org.leiyang.common.entities.MsgOneToOne;
import org.leiyang.common.entities.User;
import org.leiyang.common.holder.MsgServerAndClientHolder;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.leiyang.common.dtos.MsgDTO;
import org.leiyang.repository.MsgOneToOneRepository;
import org.leiyang.service.OfflineMessageService;
import org.leiyang.service.feign.RemoteService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static org.leiyang.common.util.Constants.MSG_SER_RSOCKET_INFO;
import static org.leiyang.common.util.Constants.OFFLINE_MSGS_KEY;

@Service
public class OfflineMessageServiceImpl implements OfflineMessageService {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final ReactiveRedisCache redisCache;
    private final MsgServerAndClientHolder msgServerAndClientHolder;
    private final MsgOneToOneRepository msgOneToOneRepository;
    private final RemoteService remoteService;
    private final RemoteCallImpl remoteCall;

    public OfflineMessageServiceImpl(ReactiveRedisCache redisCache, MsgServerAndClientHolder msgServerAndClientHolder,
                                     MsgOneToOneRepository msgOneToOneRepository, RemoteService remoteService,
                                     RemoteCallImpl remoteCall) {
        this.redisCache = redisCache;
        this.msgServerAndClientHolder = msgServerAndClientHolder;
        this.msgOneToOneRepository = msgOneToOneRepository;
        this.remoteService = remoteService;
        this.remoteCall = remoteCall;
    }

    @Override
    public Mono<R<List<MsgDTO>>> listAllHistoryMsgs(Long currentUid, Long toUid, Integer currentPage) {
        Flux<MsgOneToOne> historyMsgsFlux = msgOneToOneRepository.listAllHistoryMsgs(currentUid, toUid);
        Mono<Map<Long, User>> usersFlux = remoteService.listUsersByIds(currentUid + "," + toUid).collectMap(User::getId);
        return historyMsgsFlux
                .flatMap(msgOneToOne ->
                        usersFlux.map(userMap -> MsgDTO.convertFromMsgOneToOne(msgOneToOne, userMap))
                )
                .sort()
                .collectList()
                .map(R::ok);
    }

    @Override
    public Mono<Boolean> offlineMsgsSendWhenLogin(Long uid, String clientIp) {
        logger.info("开始查询用户({})客户端地址: ({}) 离线消息: ", uid, clientIp);
        final String key = OFFLINE_MSGS_KEY + uid;
        Mono<ConnectedMsgServerInfo> msgServersFlux = msgServerAndClientHolder.msgServerByClientIp(uid, clientIp);
        Flux<MsgOneToOne> offlineMsgsFlux =  redisCache.getCacheSet(key).map(obj -> (MsgOneToOne) obj);
        Mono<Long> delKeyMono = redisCache.deleteObject(key);
        return offlineMsgsFlux
//            .map(Message::convertFromMsgOneToOne)
            .collectList()
            .flatMap(msgOneToOnes -> {
                String uIds = msgOneToOnes.stream()
                        .map(MsgOneToOne::getFromUserId)
                        .distinct()
                        .map(String::valueOf)
                        .collect(Collectors.joining(","));
                return remoteService.listUsersByIds(uIds)
                        .collectMap(User::getId)
                        .map(userMap -> msgOneToOnes.stream()
                                .sorted(Comparator.comparing(MsgOneToOne::getTime)) //生效排列，用于APP端离线消息的显示
                                .map(msgOneToOne -> MsgDTO.convertFromMsgOneToOne(msgOneToOne, userMap))
                                .collect(Collectors.toList())
                        );
            })
            .zipWith(msgServersFlux)
            .flatMap(tuple2 -> {
                ConnectedMsgServerInfo connectedMsgServerInfo = tuple2.getT2();
                MsgServerInfo msgServerInfo = connectedMsgServerInfo.getMsgServerInfo();
                return redisCache.getCacheObject(MSG_SER_RSOCKET_INFO.concat(msgServerInfo.getServerAddress()).concat(":").concat(msgServerInfo.getServerPort()))
                    .map(addPortObj -> {
                        if(Objects.isNull(addPortObj)) {
                            logger.error("未获取到对应的MsgServer地址信息!");
                        }
                        return String.valueOf(addPortObj);
                    })
                    .flatMap(addPort -> remoteCall.batchSendOffLineMsgs(tuple2.getT1(), connectedMsgServerInfo.getChannelId(), addPort, false));
                }
            )
            .flatMap(batchSend -> {
                logger.info("批量发送离线消息到客户端(用户ID: {})的结果: {}", uid, batchSend);
                //确保离线消息发送给了当前在线的客户端之后，即可删除该批离线消息
                if(batchSend) {
                    return delKeyMono;
                }
                return Mono.just(0L);
            })
            .map(delNum -> true);
    }
}
